package cn.com.utt.common.protorpc;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Service;

import com.google.protobuf.ByteString;
import com.google.protobuf.Message;

import cn.com.utt.common.exception.RPCException;
import cn.com.utt.common.protorpc.codec.MessageDecoder;
import cn.com.utt.common.protorpc.codec.MessageEncoder;
import cn.com.utt.common.protorpc.message.Callback;
import cn.com.utt.common.protorpc.message.RPCMessage;
import cn.com.utt.common.protorpc.proto.ArgProto;
import cn.com.utt.common.protorpc.proto.RequestProto;
import cn.com.utt.common.protorpc.proto.RequestProto.RpcRequest;
import cn.com.utt.common.protorpc.proto.ResponseProto;
import cn.com.utt.common.protorpc.proto.ResponseProto.RpcResponse;
import cn.com.utt.common.util.Token;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolHandler;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;

@Service("rpcClient")
@PropertySource(value = "classpath:application-common.properties")
public class RPCClient implements InitializingBean, DisposableBean {
	
	@Autowired
	private ServiceDiscovery serviceDiscovery;

	private NioEventLoopGroup group = new NioEventLoopGroup();
	private FixedChannelPool pool = null;
	private int maxConnections = 2;

	public void connect() {
		System.out.println("init");
		try {
			Bootstrap bootstrap = new Bootstrap();
			bootstrap.group(group).channel(NioSocketChannel.class);

			String serverAddress = serviceDiscovery.discover();
			String[] array = serverAddress.split(":");
			String host = array[0];
			int port = Integer.parseInt(array[1]);
			System.out.println("rpchost:" + host);
			bootstrap.remoteAddress(new InetSocketAddress(host, port));
			final DisconnectHandler disconnectHandler = new DisconnectHandler(this);
			pool = new FixedChannelPool(bootstrap, new AbstractChannelPoolHandler() {
				@Override
				public void channelCreated(Channel ch) throws Exception {

					ch.pipeline().addLast("messageDecoder", new MessageDecoder());
					ch.pipeline().addLast("messageEncoder", new MessageEncoder());
					ch.pipeline().addLast("idleTimeoutHandler", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS));
					ch.pipeline().addLast("heartbeatHandler", new HeartbeatHandler());
					ch.pipeline().addLast("disconnectHandler", disconnectHandler);
				}
			}, maxConnections);
			new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void shutdown() {
		System.out.println("shutdown");
		if (pool != null) {
			pool.close();
			pool = null;
		}
		if (group != null) {
			group.shutdownGracefully();
			group = null;
		}
	}

	private Message callBlockingMethod(final String serviceMethod, final Message args, Message reply, int timeout)
			throws RPCException {
		if (pool == null || group == null) {
			connect();
		}
		final Callback callback = new Callback(args, reply);
		final Future<Channel> futrue = pool.acquire();
		futrue.addListener(new FutureListener<Channel>() {
			@Override
			public void operationComplete(Future<Channel> future) throws Exception {
				if (futrue.isSuccess()) {
					Channel ch = futrue.getNow();
					MessageDecoder decoder = (MessageDecoder) ch.pipeline().get("messageDecoder");
					long nextSeqId = decoder.getNextSeqId();
					RPCMessage rpcRequest = buildRequest(serviceMethod, nextSeqId, args);
					decoder.registerCallback(nextSeqId, callback);
					ch.writeAndFlush(rpcRequest);
					pool.release(ch);
				}
			}
		});
		synchronized (callback) {
			while (!callback.isDone()) {
				try {
					if (timeout == 0) {
						callback.wait();
					} else {
						callback.wait(timeout * 1000);
						break;
					}
				} catch (InterruptedException e) {
					System.out.println("Interrupted while blocking" + e);
				}
			}
		}
		if (!callback.isDone()) {
			throw new RPCException("local timeout");
		} else if (callback.getError() != null && callback.getError().length() > 0) {
			throw new RPCException(callback.getError());
		}
		return callback.getReply();
	}

	private RPCMessage buildRequest(String serviceMethod, long seqId, Message request) {
		RPCMessage msg = new RPCMessage();
		ArgProto.Arg.Builder reqBuilder = ArgProto.Arg.newBuilder();
		reqBuilder.setSeq(seqId);
		reqBuilder.setServiceMethod(serviceMethod);
		ArgProto.Arg header = reqBuilder.build();
		msg.setHeader(header);
		msg.setBody(request);
		return msg;
	}

	public void send(String topic, cn.com.utt.common.message.Message req) throws RPCException {
		RequestProto.RpcRequest.Builder requestBuilder = RequestProto.RpcRequest.newBuilder();
		requestBuilder.setKey(Token.getToken());
		requestBuilder.setTopic(topic);
		requestBuilder.setVer(0);
		String buf = req.encode();
		requestBuilder.setMsg(ByteString.copyFrom(buf.getBytes()));
		RpcRequest request = requestBuilder.build();
		try {
			callBlockingMethod("Dispatcher.Send", request, null, 0);
		} catch (RPCException e) {
			throw new RPCException(e);
		}
	}

	public cn.com.utt.common.message.Message call(String topic, cn.com.utt.common.message.Message req)
			throws RPCException {
		RequestProto.RpcRequest.Builder requestBuilder = RequestProto.RpcRequest.newBuilder();
		requestBuilder.setKey(Token.getToken());
		requestBuilder.setTopic(topic);
		requestBuilder.setVer(0);
		String buf = req.encode();
		requestBuilder.setMsg(ByteString.copyFrom(buf.getBytes()));
		RpcRequest request = requestBuilder.build();
		ResponseProto.RpcResponse.Builder respBuilder = ResponseProto.RpcResponse.newBuilder();
		RpcResponse response = respBuilder.build();
		try {
			response = (RpcResponse) callBlockingMethod("Dispatcher.Call", request, response, 0);
		} catch (RPCException e) {
			throw new RPCException(e);
		}
		if (response == null) {
			return null;
		}

		cn.com.utt.common.message.Message resp = cn.com.utt.common.message.Message
				.NewMessage(response.getMsg().toByteArray());
		if (resp == null) {
			return null;
		}
		try {
			resp.decode(response.getMsg().toByteArray());
		} catch (Exception e) {
			throw new RPCException(e);
		}
		return resp;
	}

	public cn.com.utt.common.message.Message callWithTimeout(String topic, cn.com.utt.common.message.Message req,
			int timeout) throws RPCException {
		RequestProto.RpcRequest.Builder requestBuilder = RequestProto.RpcRequest.newBuilder();
		requestBuilder.setKey(Token.getToken());
		requestBuilder.setTopic(topic);
		requestBuilder.setVer(0);
		requestBuilder.setTimeout(timeout);
		String buf = req.encode();
		requestBuilder.setMsg(ByteString.copyFromUtf8(buf));
		RpcRequest request = requestBuilder.build();
		ResponseProto.RpcResponse.Builder respBuilder = ResponseProto.RpcResponse.newBuilder();
		RpcResponse response = respBuilder.build();
		try {
			response = (RpcResponse) callBlockingMethod("Dispatcher.CallWithTimeout", request, response, timeout);
		} catch (Exception e) {
			throw new RPCException(e);
		}
		if (response == null) {
			return null;
		}
		byte[] temp = response.getMsg().toByteArray();
		System.out.println("------test");
		System.out.println(response.getMsg().isValidUtf8());
		if(!response.getMsg().isValidUtf8()){
			try {
				temp = new String(new String(temp,"gbk").getBytes("utf-8"),"utf-8").getBytes();
			} catch (UnsupportedEncodingException e1) {
				throw new RPCException(e1);
			}
		}
		cn.com.utt.common.message.Message resp = cn.com.utt.common.message.Message
				.NewMessage(temp);
		if (resp == null) {
			return null;
		}
		try {
			resp.decode(temp);
		} catch (Exception e) {
			throw new RPCException(e);
		}
		return resp;
	}

	@Override
	public void destroy() throws Exception {
		shutdown();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		connect();
	}

}
